package com.bds.btcdc.adapter.zb;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bds.btcdc.api.model.Result;
import com.bds.btcdc.model.CoinPairExchanges;
import com.bds.btcdc.model.MarketDepth;
import com.bds.btcdc.model.Order;
import com.bds.btcdc.model.TransactionRecord;
import com.bds.btcdc.service.CoinPairExchangesService;
import com.bds.btcdc.service.TransactionRecordService;
import com.bds.btcdc.util.Const;
import com.bds.btcdc.util.SpringContextHolder;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft;
import org.java_websocket.handshake.ServerHandshake;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.simp.stomp.StompSession;

import java.net.URI;
import java.nio.ByteBuffer;
import java.text.DecimalFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by DongChenchen on 2018/4/9 0009 ZB的WebSocket客戶端，接受ZB的成交历史推送
 */
public class ZBTransactionRecordReceiver extends WebSocketClient {
	private Logger log = LoggerFactory.getLogger(getClass());
	private final static DecimalFormat df = new DecimalFormat("#0.0000");
	private static final DecimalFormat df2 = new DecimalFormat("#0.0000");
	private final static String EXCHANGE = "ZB";
	private StompSession session;
	private List<CoinPairExchanges> zb_pairs;
	private CoinPairExchangesService exchangesService;
	private TransactionRecordService transactionRecordService;

	public ZBTransactionRecordReceiver(StompSession session, URI serverURI, List<CoinPairExchanges> zb_pairs) {
		super(serverURI);
		this.session = session;
		this.zb_pairs = zb_pairs;
		this.exchangesService = SpringContextHolder.getBean(CoinPairExchangesService.class);
		this.transactionRecordService = SpringContextHolder.getBean(TransactionRecordService.class);
	}


	public ZBTransactionRecordReceiver(URI serverURI) {
		super(serverURI);
	}

	@Override
	public void onOpen(ServerHandshake handshakedata) {
		for (CoinPairExchanges pair_exchanges : zb_pairs) {
			String pair_code = pair_exchanges.getPair_code().replace("_","");
			String dataMsg = "{'event':'addChannel','channel':'" +pair_code+ "_trades'}";
			send(dataMsg);
		}
		log.info("######### ZB 获取交易历史  连接成功 #############");
	}

	@Override
	public void onClose(int code, String reason, boolean remote) {
		log.info("ZB 连接关闭,code： " + code + " 异常信息：" + reason);
//		if (code == -1||code==1006) {
			new Thread() {
				@Override
				public void run() {
					try {
						Thread.sleep(Const.RE_CONN_TIME);
						log.info("ZB 重新建立连接........");
						new ZBTransactionRecordReceiver(session, new URI(Const.ZB.WS_URL), zb_pairs).connect();
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}.start();
//		}
	}

	@Override
	public synchronized void onMessage(String message) {
		try{
//			log.info("ZB 获取交易历史...");
			Result ret = new Result();
			JSONObject json= JSON.parseObject(message);
			String channel=json.getString("channel");
			String pair = channel.split("_")[0];
			String btcPair = exchangesService.getBtcPair(Const.ZB.EXCHANGE, pair);
			JSONArray data=json.getJSONArray("data");
			List<TransactionRecord> records = data.toJavaList(TransactionRecord.class);
				records.forEach(re->{
					re.setExchange(Const.ZB.EXCHANGE);
					re.setPair(btcPair);
					String  date= new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(Long.parseLong(re.getDate())*1000L);
					re.setDate(date);
					try {
						//持久化到mysql,涨跌判定 使用
						TransactionRecord record = transactionRecordService.getByExchangeAndTid(Const.ZB.EXCHANGE, re.getTid());
						if(record==null){
							transactionRecordService.insertOne(re);
						}else{
							log.info("ZB获取交易历史 插入失败,已经存在！"+JSONObject.toJSONString(re));
						}
					}catch (Exception e){
						e.printStackTrace();
						log.info("ZB获取交易历史 插入失败！!"+JSONObject.toJSONString(re));
					}
				});

			ret.setCode(Const.SUC);
			ret.setData(records);
//			if (session != null) {
				//推到btc-web主题
//				synchronized (session) {
//				session.send(Const.ZB.RECORD_TOPIC, JSONObject.toJSONString(ret).getBytes());
//				}
//			}
		}catch (Exception e){
			e.printStackTrace();
		}
	}

	@Override
	public void onMessage(ByteBuffer message) {
		log.info("received ByteBuffer");
	}

	@Override
	public void onError(Exception ex) {
		log.info("an error occurred:" + ex);
	}

	// 测试Receiver
	public static void main(String[] args) throws Exception {
		new ZBTransactionRecordReceiver(new URI("wss://api.zb.com:9999/websocket")).connect();
	}
}